import 'dart:convert';
import 'dart:isolate';

import 'package:collection/collection.dart';
import 'package:im_app/im/socket/service/RpcSendDataService.dart';
import 'package:im_app/im/util/ImCacheFactory.dart';
import 'package:im_app/im/util/ImSingleton.dart';

import '../dao/ImMessageDao.dart';
import '../dao/ImSessionDao.dart';
import '../pojo/ImPojo.dart';
import '../socket/dto/base/Resp.dart';
import '../socket/dto/business/SendData.dart';
import '../util/LoggerUtil.dart';
import '../util/UiUtil.dart';
import 'BaseService.dart';

class ImMessageReadService extends BaseService {

  final _logger = createLogger();

  static const int CACHE_TIME = 5 * 60 * 1000;   // 缓存5分钟
  static const int BATCH_SIZE = 200;             // 未读消息众多时，采取批量操作，批量操作数量

  // 保证实例只有唯一一个
  static ImMessageReadService _MESSAGE_READ_SERVICE = ImMessageReadService._();

  List<ImMessage> _cacheMessageList = <ImMessage>[];
  Map<String, int> readHandingMap = {};
  bool _isRunning = false;   // 是否正在处理已读消息
  int addHotMessageTotal = 0;
  int readHotMessageTotal = 0;

  ImMessageReadService._() {}

  static ImMessageReadService getInstance() {
    return ImMessageReadService._MESSAGE_READ_SERVICE;
  }

  void asyHandleReadMessage(int sessionId, String userId, List<ImMessage> messages) {
    List<ImMessage> needHandleMessages = _filterNeedHandleMessages(userId, messages);
    addHotMessageTotal += needHandleMessages.length;
    _logger.i("read messages add sessionId $sessionId, size: ${needHandleMessages.length}, total: $addHotMessageTotal");
    if (needHandleMessages.isNotEmpty) {
      _cacheMessageList.addAll(needHandleMessages);
    }
    if (!_isRunning) {
      if (_cacheMessageList.isEmpty) return;
      var messages = List.of(_cacheMessageList);
      _cacheMessageList = <ImMessage>[];
      _doBusinessReadMessages(userId, messages);
    } else {
      Future.delayed(const Duration(milliseconds: 1000), (){
        if (_cacheMessageList.isEmpty) return;
        var messages = List.of(_cacheMessageList);
        _cacheMessageList = <ImMessage>[];
        _doBusinessReadMessages(userId, messages);
      });
    }
  }

  Future<void> sendReadMessages(String userId, List<ImMessage> messages) async {
    var startTime = DateTime.now();
    var messageDao = ImMessageDao();
    List<ImMessage> sendReadMessages = _filterNeedHandleMessages(userId, messages);
    if (sendReadMessages.isNotEmpty) {
      sendReadMessages.map((m) {
        m.actionStatus = ImMessageActionStatus.read;
        m.updateTime = DateTime.now();
      });
      await messageDao.batchChangeMessageStatus(sendReadMessages);

      List<String> serverIds = sendReadMessages.where((m) => m.serverId != null).map((m) => m.serverId!).toList();

      _logger.i('sendReadMessage step 1 time: ${DateTime.now().millisecond - startTime.millisecond}, ${Isolate.current.hashCode}');
      RpcMessageService rpcMessageService = RpcMessageService();
      var messageReadReq = MessageReadReq(messageIds: serverIds);
      await rpcMessageService.sendReadMessages(messageReadReq);
      _logger.i('sendReadMessage step 2 time: ${DateTime.now().millisecond - startTime.millisecond}, ${Isolate.current.hashCode}');

      sendReadMessages.forEach((m) {
        m.actionStatus = ImMessageActionStatus.readed;
        m.updateTime = DateTime.now();
      });
      await messageDao.batchChangeMessageStatus(sendReadMessages);

      _logger.i('sendReadMessage step 3 time: ${DateTime.now().millisecond - startTime.millisecond}, ${Isolate.current.hashCode}');

      Map<int, List<ImMessage>> sessionMessageMap = groupBy(sendReadMessages, (m) => m.sessionId!);
      for(var key in sessionMessageMap.keys) {
        var session = sessionMessageMap[key]![0].imSession!;
        await _reduceHotMessageTotal(userId, session, sessionMessageMap[key]!.length);
      }
      _logger.i('sendReadMessage step 4 time: ${DateTime.now().millisecond - startTime.millisecond}, ${Isolate.current.hashCode}');

      UiUtil.uiUpdateMessages(messages);

      _logger.i('sendReadMessage step 5 time: ${DateTime.now().millisecond - startTime.millisecond}, ${Isolate.current.hashCode}');
    }
  }

  void _doBusinessReadMessages(String userId, List<ImMessage> needHandleMessageList) async {
    Set<String> messageServerIdSet = new Set<String>();         // 发送已读消息的serverId集合
    Set<String> messageIdSet = new Set<String>();               // 需发送已读消息serverId集合，并改变为已读

    var startTime = DateTime.now();

    for (var message in needHandleMessageList) {
      messageServerIdSet.add(message.serverId!);
      if (message.actionStatus != ImMessageActionStatus.read) {
        messageIdSet.add(message.serverId!);
      }
      readHandingMap.putIfAbsent(message.serverId!, ()=> DateTime.now().millisecond);
    }

    if (messageServerIdSet.isNotEmpty) {
      _isRunning = true;
      try {
        await _changeActionStatus(userId, messageIdSet, needHandleMessageList);
        _logger.i('sendReadMessage step 1 time: ${DateTime.now().millisecond - startTime.millisecond}');

        await _sendReadActionToServer(userId, messageServerIdSet);

        _logger.i('sendReadMessage step 2 time: ${DateTime.now().millisecond - startTime.millisecond}');

        needHandleMessageList.forEach((m)=>m.actionStatus = ImMessageActionStatus.readed);

        Map<int, List<ImMessage>> messageMap = groupBy(needHandleMessageList, (m) => m.sessionId!);
        for (var key in messageMap.keys) {
          var size = messageMap[key]!.length;
          await _reduceHotMessageTotal(userId, messageMap[key]![0].imSession!, size);
        }

        _logger.i('sendReadMessage step 3 time: ${DateTime.now().millisecond - startTime.millisecond}');

        UiUtil.uiUpdateMessages(needHandleMessageList);

        _logger.i('sendReadMessage step 4 time: ${DateTime.now().millisecond - startTime.millisecond}');
      } catch (e) {
        _logger.e("read message sending to server failure", error: e);
      } finally {
        _isRunning = false;
        messageServerIdSet.forEach((id)=> readHandingMap.remove(id));
      }
    }
  }

  Future<void> _sendReadActionToServer(String userId, Set<String> messageServerIdSet) async {
    var rpcMessageService = RpcMessageService();
    var imMessageDao = ImMessageDao();

    MessageReadReq messageReadReq = MessageReadReq(messageIds: []);
    messageReadReq.messageIds = messageServerIdSet.toList();

    _logger.i("read messages sending to server ${jsonEncode(messageReadReq)}");
    OkResp okResp = await rpcMessageService.sendReadMessages(messageReadReq);
    if (!okResp.isSuccess()) {
      _logger.e("发送已读消息失败");
    } else {
      if (messageServerIdSet.isNotEmpty) {
        imMessageDao.changeMessageActionStatus2(userId, messageServerIdSet, ImMessageActionStatus.readed);
      }
    }
  }

  Future<void> _changeActionStatus(String userId, Set<String> changeActionStatusServerIdSet, List<ImMessage>
    changeMessageList) async {
    var imMessageDao = ImMessageDao();
    changeMessageList.forEach((m)=>m.actionStatus = ImMessageActionStatus.read);
    if (changeActionStatusServerIdSet.isNotEmpty) {
      int i = 0;
      int total = 0;
      var changeIdSet = Set<String>();
      for (var serverId in changeActionStatusServerIdSet) {
        i++;
        changeIdSet.add(serverId);
        if (i == BATCH_SIZE) {
          total += i;
          imMessageDao.changeMessageActionStatus2(userId, changeIdSet, ImMessageActionStatus.read);
          changeIdSet = Set<String>();
          i = 0;
        }
      }
      if (i > 0) {
        total += i;
        imMessageDao.changeMessageActionStatus2(userId, changeIdSet, ImMessageActionStatus.read);
      }
      _logger.i("send message to change status size $total, 应该总数： ${changeActionStatusServerIdSet.length}");
    }
  }

  Future<void> _reduceHotMessageTotal(String userId, ImSession session, int size) async {
    var sessionDao = ImSessionDao();
    readHotMessageTotal+=size;
    int total = ImCacheFactory.getInstance().incHotMessage(session.id!, -size);
    _logger.i("read messages and red size: $size, total: $total, readedTotal: $readHotMessageTotal");
    await sessionDao.decHotMessageNum(session.id!, size);
    session.hotMessageTotal = total;
    // var dbSession = await sessionDao.getById(session.id!);
    // await sessionDao.loadGroupInfo(dbSession!);
    UiUtil.uiChangeSession(session!);

    // var imSession = ImSingleton.getInstance().sessionListCubit?.getData(session.id!);
    // if (imSession != null) {
    //   imSession.hotMessageTotal = total;
    //   UiUtil.uiUpdateSession(imSession);
    // } else {
    //   var dbSession = await sessionDao.getById(session.id!);
    //   await sessionDao.loadGroupInfo(dbSession!);
    //   UiUtil.uiAddSession(dbSession!);
    // }
  }

  List<ImMessage> _filterNeedHandleMessages(String userId, List<ImMessage> messages) {
    List<ImMessage> needHandleMessages = [];    // 改变状态的为已读的消息
    for (var message in messages) {
      if (message.isSender(userId)) continue;
      if (message.sessionId == null) continue;
      if (message.actionStatus != null && message.actionStatus != ImMessageActionStatus.none && message.actionStatus
          != ImMessageActionStatus.read) continue;

      if (message.isAttachFile() && message.fileStatus != ImMessageFileStatus.downloaded) continue;

      if (readHandingMap.containsKey(message.serverId)) {
        var time = readHandingMap[message.serverId]!;
        if (DateTime.now().millisecond - time < CACHE_TIME) {
          continue;
        }
      }
      needHandleMessages.add(message);
    }
    return needHandleMessages;
  }
}